package com.xhwl.logistics.service.impl.kafka;

import com.alibaba.fastjson.JSON;
import com.xhwl.logistics.dto.kafka.VehicleAlarmDTO;
import com.xhwl.logistics.service.impl.redis.VehicleExceptionRedisServiceImpl;
import com.xhwl.logistics.service.impl.redis.VehicleLogRedisServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Async;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.Date;

//@Service
public class VehicleAlarmKafkaServiceImpl {

    private static final String alarmTopic = "alarm_topic";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private VehicleExceptionRedisServiceImpl vehicleExceptionRedisService;

    @Autowired
    private VehicleLogRedisServiceImpl vehicleLogRedisService;

    @Async("threadPoolTaskExecutorForAlarm")
    public void sendAlarmForVehicleDeviceStatusOffline(Integer projectId, Integer deviceId) {
        String alarmCodeForVehicleDeviceStatusOffline = "93296064";

        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(alarmTopic, JSON.toJSONString(new VehicleAlarmDTO(projectId, alarmCodeForVehicleDeviceStatusOffline, deviceId, new Date())));

        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            final String key = "send-alarm-for-device-status";
            @Override
            public void onFailure(Throwable ex) {
                vehicleExceptionRedisService.exceptionMessage(key, ex.getMessage());
            }
            @Override
            public void onSuccess(SendResult<String, String> result) {
                vehicleLogRedisService.logWithTime(key, "sendAlarmForVehicleDeviceStatusOffline success:" + result);
            }
        });
    }

    public void sendAlarmForVehicleCarInExceptionPass(Integer projectId, Integer deviceId, String imgUrl) {
        String alarmCodeForVehicleCarInExceptionPass = "22278328";

        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(alarmTopic, JSON.toJSONString(new VehicleAlarmDTO(projectId, alarmCodeForVehicleCarInExceptionPass, deviceId, new Date(), imgUrl)));

        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            final String key = "send-alarm-for-car-in-exception-pass";
            @Override
            public void onFailure(Throwable ex) {
                vehicleExceptionRedisService.exceptionMessage(key, ex.getMessage());
            }
            @Override
            public void onSuccess(SendResult<String, String> result) {
                vehicleLogRedisService.logWithTime(key, "sendAlarmForVehicleCarInExceptionPass success:" + result);
            }
        });
    }

    public void sendAlarmForVehicleCarOutExceptionPass(Integer projectId, Integer deviceId, String imgUrl) {
        String alarmCodeForVehicleCarOutExceptionPass = "74576294";

        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(alarmTopic, JSON.toJSONString(new VehicleAlarmDTO(projectId, alarmCodeForVehicleCarOutExceptionPass, deviceId, new Date(), imgUrl)));

        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            final String key = "send-alarm-for-car-out-exception-pass";
            @Override
            public void onFailure(Throwable ex) {
                vehicleExceptionRedisService.exceptionMessage(key, ex.getMessage());
            }
            @Override
            public void onSuccess(SendResult<String, String> result) {
                vehicleLogRedisService.logWithTime(key, "sendAlarmForVehicleCarOutExceptionPass success:" + result);
            }
        });
    }

    public void sendAlarmForVehicleBlacklist(Integer projectId, Integer deviceId, String imgUrl, String content) {
        String alarmCodeForVehicleBlacklist = "70594442";

        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(alarmTopic, JSON.toJSONString(new VehicleAlarmDTO(projectId, alarmCodeForVehicleBlacklist, deviceId, new Date(), imgUrl, content)));

        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            final String key = "send-alarm-for-plate-number-blacklist";
            @Override
            public void onFailure(Throwable ex) {
                vehicleExceptionRedisService.exceptionMessage(key, ex.getMessage());
            }
            @Override
            public void onSuccess(SendResult<String, String> result) {
                vehicleLogRedisService.logWithTime(key, "sendAlarmForVehicleBlacklist success:" + result);
            }
        });
    }

}
